package com.study.redis.stream.consumer.listener;

import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;

/**
 * @author ljt
 * @description:
 * @date 2024/5/21 11:30
 */
@Slf4j
@Component
public class MessageStreamListener implements StreamListener<String, MapRecord<String, String, String>> {
    @Resource
    private RedisTemplate redisTemplate;

    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        //拿到当前组名
        String group = Thread.currentThread().getName().substring(0,Thread.currentThread().getName().lastIndexOf("-"));
        //拿到当前消费者名
        String consumer = Thread.currentThread().getThreadGroup().getName();
        // 消息ID
        RecordId messageId = message.getId();
        // 消息的key和value
        Map<String, String> body = message.getValue();
        if ("abc".equals(body.get("key"))){
            // 后续操作...
            log.error("组:[{}], 消费者:[{}], 不做数据处理", group, consumer);
        }else {
            log.info("组:[{}], 消费者:[{}], 消息: messageId:[{}], stream:[{}], body:[{}]", group,consumer,messageId, message.getStream(), body);
        }
        // 通过RedisTemplate手动确认消息
        this.redisTemplate.opsForStream().acknowledge("key", message);
    }
}
